package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class Demo2Filter {
    private HConnection connection;

    private HBaseAdmin hBaseAdmin;

    private TableName tableName = TableName.valueOf("shujia:student");

    @Before
    public void init() {
        // 1、创建连接

        Configuration configuration = new Configuration();

        /*
         * 不需要执行hmaster和regionserver的地址m  hclient会自动去zookeeper中获取，
         * 由于获取到的是一个域名，所以本地必须配置hosts
         *
         */

        //设置zookeeper地址
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {

            //创建连接  对数据增删改查使用
            connection = HConnectionManager.createConnection(configuration);

            /*
             * 创建修改删除表使用
             *
             */
            hBaseAdmin = new HBaseAdmin(connection);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 不适用过滤器, 所有数据传输到客户端进行过滤, 网络压力较大
     */
    @Test
    public void filterGender() {

        try {
            HTableInterface table = connection.getTable(tableName);

            Scan scan = new Scan();

            //全表扫描
            ResultScanner scanner = table.getScanner(scan);

            Result next;

            while ((next = scanner.next()) != null) {

                String gender = Bytes.toString(next.getValue("info".getBytes(), "gender".getBytes()));

                if ("男".equals(gender)) {

                    String name = Bytes.toString(next.getValue("info".getBytes(), "name".getBytes()));
                    String age = Bytes.toString(next.getValue("info".getBytes(), "age".getBytes()));
                    String clazz = Bytes.toString(next.getValue("info".getBytes(), "clazz".getBytes()));

                    String id = Bytes.toString(next.getRow());

                    System.out.println(id + "\t" + name + "\t" + age + '\t' + gender + '\t' + clazz);
                }
            }


        } catch (IOException e) {
            e.printStackTrace();
        }


    }

    /**
     * 列值过滤器
     */


    @Test
    public void fIlterClazz() {

        /**
         * 取出文科的学生
         *
         */

        try {
            HTableInterface table = connection.getTable(tableName);


            Scan scan = new Scan();


            //创建比较器

            //正则比较器
//            RegexStringComparator regexStringComparator = new RegexStringComparator("文科*");

            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("文科".getBytes());


            //列值过滤器
            //列值过滤器也需要做全表扫描, 但是传递给客户端的数据量便少了
            SingleColumnValueFilter columnValueFilter = new SingleColumnValueFilter(
                    "info".getBytes(),//列簇
                    "clazz".getBytes(),//列名
                    CompareFilter.CompareOp.EQUAL,//比较方式
                    binaryPrefixComparator//比较器
            );

            //增加过滤器
            scan.setFilter(columnValueFilter);


            //执行扫描
            ResultScanner scanner = table.getScanner(scan);


            //打印结果
            print(scanner);

        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    /**
     * rowkey过滤器
     */

    @Test
    public void filterRowkey() {

        //取出id以  15001001    开头的学生
        try {
            HTableInterface table = connection.getTable(tableName);


            Scan scan = new Scan();
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("15001001".getBytes());

            //行键过滤器  效率最高的,  因为rowkey是有序的不需要做全表扫描
            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

            //增加过滤器
            scan.setFilter(rowFilter);

            ResultScanner scanner = table.getScanner(scan);

            print(scanner);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public static void print(ResultScanner scanner) throws IOException {
        Result next;

        while ((next = scanner.next()) != null) {

            String gender = Bytes.toString(next.getValue("info".getBytes(), "gender".getBytes()));
            String name = Bytes.toString(next.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(next.getValue("info".getBytes(), "age".getBytes()));
            String clazz = Bytes.toString(next.getValue("info".getBytes(), "clazz".getBytes()));
            String id = Bytes.toString(next.getRow());

            System.out.println(id + "\t" + name + "\t" + age + '\t' + gender + '\t' + clazz);
        }
    }

}
